package com.zz.tijian.handle.mq;
import com.zz.tijian.service.OrdersService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.utils.SerializationUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Service
@Slf4j
public class RefundConsumer {

    @Autowired
    private OrdersService ordersService;

    private static final int MAX_RETRY_ATTEMPTS = 3; // 设置最大重试次数
    private Map<Integer, Integer> counterMap=new ConcurrentHashMap<>();

    @RabbitListener(queues = RabbitMQConfig.REFUND_QUEUE)
    public void handleRefundMessage(Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
        Integer orderId = null;
        orderId = (Integer) SerializationUtils.deserialize(message.getBody()); // 反序列化消息体获取订单号
        Integer integer = counterMap.get(orderId);
        if (integer==null){
            counterMap.put(orderId, 1);
        }else{
            counterMap.put(orderId,integer.intValue() + 1);
        }

        try {
            ordersService.processRefund(orderId);
            log.info("Refund processing successful for order {}", orderId);
            counterMap.remove(orderId);
            channel.basicAck(deliveryTag, false); // 成功处理后确认消息
        } catch (Exception e) {
            log.error("出现异常: {}", orderId, e);
            if (counterMap.get(orderId) >= MAX_RETRY_ATTEMPTS) {
                // 超过最大重试次数，拒绝消息且不重新入队
                channel.basicNack(deliveryTag, false, false);
                log.info("订单-{}退款失败,向死信队列发送信息", orderId);
                counterMap.remove(orderId);
            } else {
                // 重试次数未到最大值，重新入队
                channel.basicNack(deliveryTag, false, true);
            }
        }
    }

    private void handleFailure(Message message, Channel channel, long deliveryTag, Integer orderId) {
        int retryCount = (int) message.getMessageProperties().getHeaders().getOrDefault("retryCount", 0);
        if (retryCount < MAX_RETRY_ATTEMPTS) {
            // 增加重试次数并重新发送消息
            retryCount++;
            message.getMessageProperties().getHeaders().put("retryCount", retryCount);
            throw new AmqpRejectAndDontRequeueException("Retry processing");
        } else {
            // 达到最大重试次数，发送到死信队列
            log.info("Retry limit exceeded for message with orderId {}", orderId);
            try {
                // 手动确认消息
                channel.basicAck(deliveryTag, false);

            } catch (IOException ioException) {
                log.error("Failed to ack message for order {}", orderId, ioException);
            }
        }
    }

    //@RabbitListener(queues = RabbitMQConfig.DLX_QUEUE)
    public void handleRetryRefundMessage(Integer orderId) {
        try {
            ordersService.processRefund(orderId);
            log.info("Retry processing successful for order {}", orderId);
        } catch (Exception e) {
            log.error("Retry processing failed for order {}", orderId, e);
            // 这里可以记录日志或者进行其他处理
        }
    }
}
